package flinkdeploylogic

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
	"yunzhan/common/models"
	utils "yunzhan/common/utils"
	"yunzhan/rpc-server/internal/svc"
	agent "yunzhan/rpc-server/pb"

	"github.com/zeromicro/go-zero/core/logx"
)

type DeployFlinkStandaloneLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewDeployFlinkStandaloneLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DeployFlinkStandaloneLogic {
	return &DeployFlinkStandaloneLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

// DeployFlinkStandalone 集群部署
func (l *DeployFlinkStandaloneLogic) DeployFlinkStandalone(in *agent.DeployFlinkRequest) (*agent.DeployResponse, error) {
	if in.NodeInfo == nil {
		return &agent.DeployResponse{
			Code:    500,
			Message: "未配置节点信息, 无法执行Flink部署",
		}, errors.New("未配置节点信息, 无法执行Flink部署")
	}

	resp := &agent.DeployResponse{
		Code:    200,
		Message: fmt.Sprintf("请求已接收，正在部署 %s 到节点 %s", in.ConfigInfo.ComponentName, in.NodeInfo.Host),
	}

	var wg sync.WaitGroup

	wg.Add(1)
	go func() {
		defer wg.Done()
		packageDirectory := in.ConfigInfo.PackagePath
		rtrimDir := utils.Rtrim(packageDirectory, "/")
		logx.Infof("安装包存放目录: %s", rtrimDir)
		flinkVersion := in.ConfigInfo.ComponentName
		logx.Infof("Flink version: %s", flinkVersion)

		installPath := utils.Rtrim(in.ConfigInfo.InstallPath, "/")
		//baseInfo := fmt.Sprintf("安装包存放目录: %s, Flink version: %s, Flink安装目录: %s", rtrimDir, flinkVersion, installPath)
		step := 1

		checkPathCommand := fmt.Sprintf(`
if [ ! -d "%s" ]; then
   sudo mkdir -p "%s"
fi
`, installPath, installPath)
		step++
		logx.Infof("[Step 1/%d] %s: 执检测组件安装路径是否存在: %s", step, in.NodeInfo.Host, checkPathCommand)
		output, err := utils.ExecCommand(checkPathCommand)
		if err != nil {
			logx.Errorf("[Step 1/%d] %s: 执检测组件安装路径失败: %v, output: %s", step, in.NodeInfo.Host, err, output)
			return
		}
		logx.Infof("[Step 1/%d] %s: 执行检测组件安装路径成功", step, in.NodeInfo.Host)

		tarCommand := fmt.Sprintf("tar -xzf %s/%s -C %s", rtrimDir, flinkVersion, installPath)
		output, err = utils.ExecCommand(tarCommand)
		if err != nil {
			logx.Errorf("[Step 2/%d] %s: 解压安装包失败: %v, output: %s", step, in.NodeInfo.Host, err, output)

			return
		}
		logx.Infof("[Step 2/%d] %s: 解压安装包成功", step, in.NodeInfo.Host)

		alertmanagerHome := utils.Rtrim(flinkVersion, ".tgz")
		home := fmt.Sprintf("%s/%s", installPath, alertmanagerHome)

		// 修改组件所属用户组
		chownCommand := fmt.Sprintf("sudo chown -R %s:%s %s", in.NodeInfo.Username, in.NodeInfo.Username, home)
		output, err = utils.ExecCommand(chownCommand)
		if err != nil {
			logx.Errorf("[Step 3/%d] %s: 修改用户组失败: %v, output: %s", step, in.NodeInfo.Host, err, output)

			return
		}
		logx.Infof("[Step 3/%d] %s: 修改组件所属用户组成功", step, in.NodeInfo.Host)

		flinkServiceFile := fmt.Sprintf(`[Unit]
Description=Flink instance
After=network.target

[Service]
User=%s
Group=%s
Type=notify
ExecStart=%s/bin/grafana server \
  --homepath=%s \
  --config=%s/conf/defaults.ini

ExecReload=/bin/kill -HUP $MAINPID
Restart=on-failure
RestartSec=5
LimitNOFILE=65535

[Install]
WantedBy=multi-user_logic.target`, in.NodeInfo.Username, in.NodeInfo.Username, home, home, home)

		// 将kafka.service文件内容写入到远程节点的/etc/systemd/system/kafka_logic.services
		writeServiceFileCommand := fmt.Sprintf("echo '%s' | sudo tee /etc/systemd/system/flink.service", flinkServiceFile)
		output, err = utils.ExecCommand(writeServiceFileCommand)
		if err != nil {
			logx.Errorf("[Step 6/%d] %s: 创建flink.service文件失败: %v, output: %s", step, in.NodeInfo.Host, err, output)
			return
		}
		logx.Infof("[Step 6/%d] %s: 创建flink.service文件成功", step, in.NodeInfo.Host)

		// 重新加载systemd以确保新的服务文件被识别
		reloadSystemdCommand := fmt.Sprintf("sudo systemctl daemon-reload && sudo systemctl enable flink.service --now")
		output, err = utils.ExecCommand(reloadSystemdCommand)
		if err != nil {
			logx.Errorf("[Step 7/%d] %s: 启动flink.service服务失败: %v, output: %s", step, in.NodeInfo.Host, err, output)
			return
		}
		logx.Infof("[Step 7/%d] %s: 执行启动flink.service服务命令成功", step, in.NodeInfo.Host)

		time.Sleep(3 * time.Second)

		statusCommand := fmt.Sprintf("pgrep -f flink && exec bash")
		output, err = utils.ExecCommand(statusCommand)
		if err != nil || output == "" {
			logx.Errorf("[Step Final] %s: Flink启动失败: %v, Output: %s", in.NodeInfo.Host, err, output)
			return
		}
		logx.Infof("[Step Final] %s: Flink部署成功", in.NodeInfo.Host)

		currentIP, currentIPErr := utils.GetCurrentInternalIP()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机IP: %v", in.NodeInfo.Host, currentIPErr)
			logx.Errorf(message)
			return
		}
		currentHost, currentHostErr := utils.GetCurrentHostname()
		if err != nil {
			message := fmt.Sprintf("%s: 无法获取当前主机名: %v", in.NodeInfo.Host, currentHostErr)
			logx.Errorf(message)
			return
		}

		flinkInfo := &models.ComponentInfo{
			ComponentName:  "Flink",
			Version:        in.Version,
			NodeHost:       currentHost,
			NodeIP:         currentIP,
			ComponentRole:  "Server",
			HomePath:       home,
			DataStorageDir: "",
			Port:           string(in.FlinkConfig.Rest.Port),
			Status:         true,
			AdditionalInfo: fmt.Sprintf("config_file=%s/conf/config.yml", home),
		}

		fmt.Println(flinkInfo)
		if err != nil {
			logx.Errorf("%s 节点: 更新 Flink 角色信息失败: %v", in.NodeInfo.Host, err)
		}
		logx.Infof("%s 节点: 更新 Flink 角色信息成功 -> %s", in.NodeInfo.Host, "Server")

	}()
	wg.Wait()
	return resp, nil
}

func setDefaults(config *agent.FlinkConfig) {
	// JobManager default values
	if config.JobManager.BindHost == "" {
		config.JobManager.BindHost = "0.0.0.0" // Default bind-host
	}
	if config.JobManager.Rpc.Address == "" {
		config.JobManager.Rpc.Address = "localhost" // Default RPC address
	}
	if config.JobManager.Rpc.Port == 0 {
		config.JobManager.Rpc.Port = 6123 // Default RPC port
	}
	if config.JobManager.Memory.Size == "" {
		config.JobManager.Memory.Size = "1600m" // Default memory size
	}
	// TaskManager default values
	if config.TaskManager.BindHost == "" {
		config.TaskManager.BindHost = "0.0.0.0" // Default bind-host
	}
	if config.TaskManager.Host == "" {
		config.TaskManager.Host = "localhost" // Default host
	}
	if config.TaskManager.NumberOfTaskSlots == 0 {
		config.TaskManager.NumberOfTaskSlots = 1 // Default task slots
	}
	if config.TaskManager.Memory.Size == "" {
		config.TaskManager.Memory.Size = "1728m" // Default memory size
	}
}
